package cc.siyecao.fastdfs.util;

import cc.siyecao.fastdfs.command.Command;
import cc.siyecao.fastdfs.model.Metadata;
import cc.siyecao.fastdfs.model.RecvHeaderInfo;
import cc.siyecao.fastdfs.model.RecvPackageInfo;

import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.net.Socket;
import java.util.*;

import static cc.siyecao.fastdfs.protocol.ProtocolConstants.*;

public class ProtocolUtil implements Command {


    /**
     * pack header by FastDFS transfer protocol
     *
     * @param cmd    which command to send
     * @param pkgLen package body length
     * @param errno  status code, should be (byte)0
     * @return packed byte buffer
     */
    public static byte[] packHeader(byte cmd, long pkgLen, byte errno) throws UnsupportedEncodingException {
        byte[] header;
        byte[] hexLen;

        header = new byte[FDFS_PROTO_PKG_LEN_SIZE + 2];
        Arrays.fill( header, (byte) 0 );

        hexLen = BytesUtil.long2buff( pkgLen );
        System.arraycopy( hexLen, 0, header, 0, hexLen.length );
        header[PROTO_HEADER_CMD_INDEX] = cmd;
        header[PROTO_HEADER_STATUS_INDEX] = errno;
        return header;
    }

    /**
     * receive pack header
     *
     * @param in            input stream
     * @param expectCmd     expect response command
     * @param expectBodyLen expect response package body length
     * @return RecvHeaderInfo: errno and pkg body length
     */
    public static RecvHeaderInfo recvHeader(InputStream in, byte expectCmd, long expectBodyLen) throws IOException {
        byte[] header;
        int bytes;
        long pkgLen;

        header = new byte[FDFS_PROTO_PKG_LEN_SIZE + 2];

        if ((bytes = in.read( header )) != header.length) {
            throw new IOException( "recv package size " + bytes + " != " + header.length );
        }

        if (header[PROTO_HEADER_CMD_INDEX] != expectCmd) {
            throw new IOException( "recv cmd: " + header[PROTO_HEADER_CMD_INDEX] + " is not correct, expect cmd: " + expectCmd );
        }

        if (header[PROTO_HEADER_STATUS_INDEX] != 0) {
            return new RecvHeaderInfo( header[PROTO_HEADER_STATUS_INDEX], 0 );
        }

        pkgLen = BytesUtil.buff2long( header, 0 );
        if (pkgLen < 0) {
            throw new IOException( "recv body length: " + pkgLen + " < 0!" );
        }

        if (expectBodyLen >= 0 && pkgLen != expectBodyLen) {
            throw new IOException( "recv body length: " + pkgLen + " is not correct, expect length: " + expectBodyLen );
        }

        return new RecvHeaderInfo( (byte) 0, pkgLen );
    }

    /**
     * receive whole pack
     *
     * @param in            input stream
     * @param expectCmd     expect response command
     * @param expectBodyLen expect response package body length
     * @return RecvPackageInfo: errno and reponse body(byte buff)
     */
    public static RecvPackageInfo recvPackage(InputStream in, byte expectCmd, long expectBodyLen) throws IOException {
        RecvHeaderInfo header = recvHeader( in, expectCmd, expectBodyLen );
        if (header.errno != 0) {
            return new RecvPackageInfo( header.errno, null );
        }

        byte[] body = new byte[(int) header.bodyLen];
        int totalBytes = 0;
        int remainBytes = (int) header.bodyLen;
        int bytes;

        while (totalBytes < header.bodyLen) {
            if ((bytes = in.read( body, totalBytes, remainBytes )) < 0) {
                break;
            }

            totalBytes += bytes;
            remainBytes -= bytes;
        }

        if (totalBytes != header.bodyLen) {
            throw new IOException( "recv package size " + totalBytes + " != " + header.bodyLen );
        }

        return new RecvPackageInfo( (byte) 0, body );
    }

    /**
     * split metadata to name value pair array
     *
     * @param metaBuff metadata
     * @return name value pair array
     */
    public static Set<Metadata> splitMetadata(String metaBuff) {
        return splitMetadata( metaBuff, FDFS_RECORD_SEPERATOR, FDFS_FIELD_SEPERATOR );
    }

    /**
     * split metadata to name value pair array
     *
     * @param metaBuff        metadata
     * @param recordSeperator record/row seperator
     * @param filedSeperator  field/column seperator
     * @return name value pair array
     */
    public static Set<Metadata> splitMetadata(String metaBuff, String recordSeperator, String filedSeperator) {
        String[] rows;
        String[] cols;
        Set<Metadata> metaList;

        rows = metaBuff.split( recordSeperator );
        metaList = new HashSet<Metadata>( rows.length );
        for (int i = 0; i < rows.length; i++) {
            cols = rows[i].split( filedSeperator, 2 );
            Metadata metadata = new Metadata( cols[0] );
            if (cols.length == 2) {
                metadata.setValue( cols[1] );
            }
            metaList.add( metadata );
        }

        return metaList;
    }

    /**
     * pack metadata array to string
     *
     * @param metaSet metadata array
     * @return packed metadata
     */
    public static String packMetadata(Set<Metadata> metaSet) {
        if (metaSet.size() == 0) {
            return "";
        }
        List<Metadata> metaList = new ArrayList<>( metaSet );
        StringBuffer sb = new StringBuffer( 32 * metaList.size() );
        sb.append( metaList.get( 0 ).getName() ).append( FDFS_FIELD_SEPERATOR ).append( metaList.get( 0 ).getValue() );
        for (int i = 1; i < metaList.size(); i++) {
            sb.append( FDFS_RECORD_SEPERATOR );
            sb.append( metaList.get( i ).getName() ).append( FDFS_FIELD_SEPERATOR ).append( metaList.get( i ).getValue() );
        }

        return sb.toString();
    }

    /**
     * send quit command to server and close socket
     *
     * @param sock the Socket object
     */
    public static void closeSocket(Socket sock) throws IOException {
        byte[] header;
        header = packHeader( FDFS_PROTO_CMD_QUIT, 0, (byte) 0 );
        sock.getOutputStream().write( header );
        sock.close();
    }

    /**
     * send ACTIVE_TEST command to server, test if network is ok and the server is alive
     *
     * @param sock the Socket object
     */
    public static boolean activeTest(Socket sock) throws IOException {
        byte[] header;
        header = packHeader( FDFS_PROTO_CMD_ACTIVE_TEST, 0, (byte) 0 );
        sock.getOutputStream().write( header );

        RecvHeaderInfo headerInfo = recvHeader( sock.getInputStream(), TRACKER_PROTO_CMD_RESP, 0 );
        return headerInfo.errno == 0 ? true : false;
    }

}
